package 生产者;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * Kafka生产者的缓冲区主要用于暂存发送到Kafka集群的消息，直到这些消息能够被发送到Kafka的broker上。这个机制有几个关键作用：
 *
 * 提高吞吐量：通过批量处理消息，生产者可以将多个小的消息合并成一个大的消息批次(batch)发送给Kafka，从而减少网络请求的数量，提高吞吐量。
 * 平滑突发流量：在网络不稳定或broker暂时不可用的情况下，缓冲区可以临时存储消息，防止生产者因短暂的网络抖动或broker负载过高而丢失消息或者抛出异常。
 * 异步发送支持：由于消息是首先被写入缓冲区而不是立即发送出去，这使得生产者可以实现高效的异步发送逻辑，允许应用程序继续执行而不必等待每条消息的确认。
 * 在配置Kafka生产者时，可以通过几个参数来调整与缓冲区相关的设置：
 *
 * buffer.memory：控制生产者缓冲区的总大小（以字节为单位）。如果消息产生的速度超过了它们能被发送的速度，导致缓冲区满了，那么进一步的send调用将会被阻塞，直到有足够空间为止。
 * batch.size：指定当数据积累到特定字节数时，生产者会尝试将一批消息打包并发送出去。适当增加这个值可以提升吞吐量，但也会稍微增加延迟。
 * linger.ms：这是一个额外的等待时间，即使达到了batch.size，生产者也会等待最多linger.ms的时间看是否还有更多的消息加入当前批次，这样可以更有效地利用网络带宽。
 * 通过合理配置这些参数，可以在性能和资源使用之间找到一个平衡点，确保既能够高效地发送大量消息，又能保持系统的稳定性和可靠性。
 */
public class KafkaExampleProducer {

    public static void main(String[] args) {
        // 配置生产者的属性
        Properties props = new Properties();
        // Kafka集群的地址
        props.put("bootstrap.servers", "localhost:9092");
        // 消息key和value的序列化方式
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 设置acks为'all'以保证消息发送成功
        props.put("acks", "all");

        // 创建Kafka生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            for (int i = 0; i < 10; i++) {
                // 创建要发送的消息记录
                ProducerRecord<String, String> record = new ProducerRecord<>("your_topic_name", "key" + i, "value" + i);
                // 发送消息并获取Future对象
                Future<RecordMetadata> future = producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        // 如果没有异常，打印消息的偏移量、分区等信息
                        System.out.println("Message sent to topic:" + metadata.topic() +
                                ", partition:" + metadata.partition() +
                                ", offset:" + metadata.offset());
                    } else {
                        // 处理发送消息时可能发生的异常
                        exception.printStackTrace();
                    }
                });

                // 可以选择同步等待消息发送完成
                RecordMetadata recordMetadata = future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者，释放资源
            producer.close();
        }
    }
}